package com.ultrapower.request;

import java.util.List;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

import com.ultrapower.bean.AgentEventBean;
import com.ultrapower.hbase.HbaseUtil;
import com.ultrapower.mapper.AgentEventMapper;
import com.ultrapower.util.BytesUtils;
import com.ultrapower.util.HBaseTables;
import com.ultrapower.util.RowKeyUtils;

/**
 * 
 * @author wangy
 *
 */
public class AgentEventTable extends BaseRequest {
	
	private int SCANNER_CACHE_SIZE = 32;

	public static void main(String[] args) {
		AgentEventTable event = new AgentEventTable();
		event.execute();
	}


	@Override
	void execute() {
		
		TableName agentEventTableName = HbaseUtil.getTableName(HBaseTables.AGENT_EVENT_STR);
		 
		this.getHbaseTemplate().find(agentEventTableName, createScan(), (ResultScanner scanner) -> {

			AgentEventMapper map = new AgentEventMapper();
			scanner.forEach(r -> {
				try {
					List<AgentEventBean> info = map.mapRow(r, 0);
					info.forEach(e->{
						System.out.println(e);
					});
				} catch (Exception e) {
					e.printStackTrace();
				}
			});
			return null;
		});
	}
	
	  private byte[] createRowKey(String agentId, long timestamp) {
	        byte[] agentIdKey = BytesUtils.toBytes(agentId);
	        long reverseTimestamp = HbaseUtil.reverseTimeMillis(timestamp);
	        return RowKeyUtils.concatFixedByteAndLong(agentIdKey, HBaseTables.AGENT_NAME_MAX_LEN, reverseTimestamp);
	    }

	@Override
	protected Scan createScan() {
		from = from/(1000*60*10)*(1000*60*10);
		Scan scan = new Scan();
		scan.setMaxVersions(1);
		scan.setCaching(SCANNER_CACHE_SIZE);
		scan.setStartRow(createRowKey(agentId, to));
		scan.setStopRow(createRowKey(agentId, from ));
		scan.addFamily(HBaseTables.AGENT_EVENT_CF_EVENTS);
		return scan;
	}

}
